package com.bd08.stock.connector;


import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.avro.Schema.Field;
import org.apache.avro.Schema.Type;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.avro.SchemaParseException;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.connect.data.Struct;

import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;

import org.apache.avro.Schema.Parser;

import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

import java.util.*;

import static org.apache.avro.Schema.Type.NULL;


import org.apache.kafka.connect.data.Schema;

/**
 * @author seal
 * @date 2022-03-11
 */
public class FinanceSourceTask extends SourceTask {

    private String topic;
    private String symbols;
    private int interval;
    private String spoofFlag;
    private String schemaUri;
    //模式定义
    private Schema dataSchema;

    private String cuid;

    @Override
    public String version() {
        return AppInfoParser.getVersion();
    }

    /**
     * 获取 connector 中 得到的 taskConfigs
     * @param props
     * @author 卫泽辉
     */
    @Override
    public void start(Map<String, String> props) {
        //从FSC中获取相应内容
        topic = props.get(FinanceSourceConnector.TOPIC_CONFIG);
        symbols = props.get(FinanceSourceConnector.STOCK_SYMBOLS_CONFIG);
        interval = Integer.parseInt(props.get(FinanceSourceConnector.REFRESH_INTERVAL_CONFIG)) * 1000;
        spoofFlag = props.get(FinanceSourceConnector.SPOOF_FLAG_CONFIG);
        cuid = props.get(FinanceSourceConnector.CUID);
        //调用自定方法构建模式
        dataSchema=getBuildSchema(SCHEMA);

    }



    /**
     * 拉取的数据封装成 记录列表
     * @return
     * @throws InterruptedException
     */
    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        ArrayList<SourceRecord> records = new ArrayList<SourceRecord>();

        for (String symbol : symbols.split(",")) {
            String stockMsg = spoofFlag.equalsIgnoreCase("none") ?
                    YahooFinanceStockHelper.getStockJson(symbols, true) :
                    YahooFinanceStockHelper.getFakedStockJson(symbol, spoofFlag);
            records.add(new SourceRecord(offsetKey(symbol),
                    offsetValue(new JSONObject(stockMsg).getString("refresh_time")),
                    topic,
                    dataSchema, structDecodingFromJson(stockMsg)));

        }
        Thread.sleep(interval);
        return records;
    }

    /**
     * 停止
     */
    @Override
    public void stop() {

    }


    private SchemaBuilder getInnerBuilder(Field field) {
        return getInnerBuilder(field, field.schema().getType());

    }

    private SchemaBuilder getInnerBuilder(Field field, Type type) {
        boolean hasDefault = field.defaultVal() != null;
        SchemaBuilder innerBuilder = null;
        switch (type) {
            case STRING: {
                innerBuilder = SchemaBuilder.string();
                if (hasDefault)


                    //原
                    //innerBuilder.defaultValue(field.defaultValue().asText());
                    //改
                    innerBuilder.defaultValue(field.defaultVal());
                break;
            }
            case INT: {
                innerBuilder = SchemaBuilder.int32();
                if (hasDefault)
                    innerBuilder.defaultValue(field.defaultVal());
                break;
            }
            case LONG: {
                innerBuilder = SchemaBuilder.int64();
                if (hasDefault)
                    innerBuilder.defaultValue(field.defaultVal());
                break;
            }
            case DOUBLE: {
                innerBuilder = SchemaBuilder.float64();
                if (hasDefault)
                    innerBuilder.defaultValue(field.defaultVal());
                break;
            }
            case FLOAT:{
                innerBuilder = SchemaBuilder.float32();
                if (hasDefault)
                    innerBuilder.defaultValue(field.defaultVal());
                break;
            }
            case BOOLEAN: {
                innerBuilder = SchemaBuilder.bool();
                if (hasDefault)
                    innerBuilder.defaultValue(field.defaultVal());
                break;
            }
            case UNION: {
                //类似与 java.util.Date 和 java.sql.Date 问题 不同组织中相同类目 通过全类名解决
                for (org.apache.avro.Schema schema : field.schema().getTypes()) {
                    if (!schema.getType().equals(NULL))
                        return getInnerBuilder(field, schema.getType());
                }
            }
            default:
                throw new ConnectException(
                        "Unable to build schema because there is no case for type " + field.schema().getType());
        }
        return innerBuilder;
    }








    public static final String KEY_FIELD = "symbol";

    private Map<String, String> offsetKey(String keyName) {
        return Collections.singletonMap(KEY_FIELD, keyName);
    }



    public static final String VALUE_FIELD = "refresh_time";
    private Map<String, String> offsetValue(String value) {
        return Collections.singletonMap(VALUE_FIELD, value);
    }



    public Struct structDecodingFromJson(String line) {

        if (line.length() > 0) {
            JsonNode json = null;
            try {
                json = new ObjectMapper().readValue(line, JsonNode.class);
            } catch (IOException ex) {
                throw new ConnectException(String.format("Unable to parse %s into a valid JSON"), ex);
            }

            Struct struct = new Struct(dataSchema);
            Iterator<Map.Entry<String, JsonNode>> iterator = json.fields();
            while (iterator.hasNext()) {
                Map.Entry<String, JsonNode> entry = iterator.next();
                Object value = null;
                org.apache.kafka.connect.data.Field theField = dataSchema.field(entry.getKey());
                if (theField != null) {
                    switch (theField.schema().type()) {
                        case STRING:
                        case BYTES: {
                            value = entry.getValue().asText();
                            break;
                        }
                        case INT32: {
                            value = entry.getValue().asInt();
                            break;
                        }
                        case INT64: {
                            value = entry.getValue().asLong();
                            break;
                        }
                        case FLOAT32:
                        case FLOAT64: {
                            value = entry.getValue().asDouble();
                            break;
                        }
                        case BOOLEAN: {
                            value = entry.getValue().asBoolean();
                            break;
                        }
                        default:
                            value = entry.getValue().asText();
                    }
                }
                System.out.println("STRUCT PUT -" + entry.getKey() + ":" + value);
                struct.put(entry.getKey(), value);

            }
            return struct;
        }
        return null;
    }


    public static final String SCHEMA = "{\"type\":\"record\"," +
            "\"name\":\"Stock\"," +
            "\"namespace\":\"com.bd08.stock.avro\"," +
            "\"fields\":[{\"name\":\"refresh_time\",\"type\":\"string\"},{\"name\":\"symbol\",\"type\":\"string\"},{\"name\":\"company_name\",\"type\":\"string\"},{\"name\":\"exchange\",\"type\":\"string\"},{\"name\":\"open_price\",\"type\":\"double\"},{\"name\":\"ask_price\",\"type\":\"double\"},{\"name\":\"ask_size\",\"type\":\"int\"},{\"name\":\"bid_price\",\"type\":\"double\"},{\"name\":\"bid_size\",\"type\":\"int\"},{\"name\":\"price\",\"type\":\"double\"}]}";


    /**
     * 构建模式
     * 通过模式字符串构建
     * @param schemaString
     * @return
     */
    private Schema getBuildSchema(String schemaString) {
        org.apache.avro.Schema avroSchema = null;
        try {
            //解析字符串构建模式
            avroSchema = new Parser().parse(schemaString);
        } catch (SchemaParseException ex) {
            throw new ConnectException(
                    String.format("Unable to successfully parse schema from: %s", schemaString), ex);
        }
        //定构建器
        SchemaBuilder builder = SchemaBuilder.struct();
        builder.name(avroSchema.getFullName());
        builder.doc(avroSchema.getDoc());
        for (org.apache.avro.Schema.Field field : avroSchema.getFields()) {
            boolean hasDefault = field.defaultVal()!=null;

            SchemaBuilder innerBuilder = getInnerBuilder(field);

            if (hasDefault)
                innerBuilder.optional();

            builder.field(field.name(), innerBuilder.build());
        }
        return builder.build();
    }




    private static final Logger log = LoggerFactory.getLogger(FinanceSourceTask.class);



}
